package com.chukun.flink.table.api.sql;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/**
 * @author chukun
 * @version 1.0.0
 * @description sql基本使用流程
 * @createTime 2022年05月29日 23:08:00
 */
public class StreamTableSqlBasic {

    public static void main(String[] args) throws Exception {

        // 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 获取表环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 加载原始数据
        DataStreamSource<String> dataStream = env.fromElements("Alice", "Bob", "John");

        // 原始数据转换为临时表
        Table origalTable = tableEnv.fromDataStream(dataStream).as("name");

        // 原始数据注册为临时表
        tableEnv.createTemporaryView("simple_table", origalTable);

        // sql查询
        Table resultTable = tableEnv.sqlQuery("SELECT UPPER(name) FROM simple_table");

        // 结果转为 stream
        DataStream<Row> resultStream = tableEnv.toDataStream(resultTable);

        resultStream.print();

        env.execute();

    }
}
